package io.xxx.xbutler.core.event.listen;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.hy.corecode.idgen.WFGIdGenerator;
import io.xxx.xbutler.client.StoreClient;
import io.xxx.xbutler.common.RedisKeys;
import io.xxx.xbutler.constant.Formats;
import io.xxx.xbutler.constant.Topics;
import io.xxx.xbutler.core.alarm.Alarm;
import io.xxx.xbutler.core.check.CheckError;
import io.xxx.xbutler.core.check.Checker;
import io.xxx.xbutler.core.get.ContentGetter;
import io.xxx.xbutler.core.select.robot.RobotSelector;
import io.xxx.xbutler.data.*;
import io.xxx.xbutler.domain.Robot;
import io.xxx.xbutler.domain.RoomCategory;
import io.xxx.xbutler.domain.Store;
import io.xxx.xbutler.domain.task.*;
import io.xxx.xbutler.sdk.TanJingClient;
import io.xxx.xbutler.sdk.domain.GroupChatMessage;
import io.xxx.xbutler.sdk.domain.PrivateChatMessage;
import io.xxx.xbutler.sdk.request.SendChatMessagesRequest;
import io.xxx.xbutler.sdk.request.SendGroupChatMessagesRequest;
import io.xxx.xbutler.sdk.request.SendPrivateChatMessagesRequest;
import io.xxx.xbutler.sdk.response.SendChatMessagesResponse;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.commons.lang3.tuple.Triple;
import org.redisson.api.RAtomicLong;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.pulsar.annotation.PulsarListener;
import org.springframework.pulsar.listener.AckMode;
import org.springframework.stereotype.Component;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StopWatch;
import org.springframework.util.StringUtils;
import org.springframework.web.util.UriComponentsBuilder;

import javax.annotation.CheckReturnValue;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;

@Slf4j
@Component
public class MessagePushListener {

    @Resource
    private List<ContentGetter> contentGetters;

    @Resource
    private Map<String, Checker> checkers;

    @Resource
    private List<Alarm> alarms;

    @Resource
    private TransactionTemplate transactionTemplate;

    @Resource
    private TaskMapper taskMapper;

    @Resource
    private TaskItemMapper taskItemMapper;

    @Resource
    private TaskReceiverMapper taskReceiverMapper;

    @Resource
    private RobotSelector robotSelector;

    @Resource
    private PushRecordMapper pushRecordMapper;

    @Resource
    private PushRecordItemMapper pushRecordItemMapper;

    @Resource
    private WFGIdGenerator idGenerator;

    @Resource
    private TanJingClient tanJingClient;

    @Resource
    private StoreClient storeClient;

    @Resource
    private RedissonClient redissonClient;

    @Value("${task.push.room-checkers:CustomerRoom:taskChecker,storeChecker;MaterialRoom:taskChecker}")
    private String roomCheckers;

    @Value("${task.push.contact-checkers:productChecker,materialChecker}")
    private String contactCheckers;

    @PulsarListener(topics = Topics.MESSAGE_PUSH, ackMode = AckMode.RECORD)
    public void received(Long taskReceiverId) {
        try {
            StopWatch watch = new StopWatch();
            watch.start();

            TaskReceiver taskReceiver = taskReceiverMapper.selectById(taskReceiverId);
            if (taskReceiver == null) {
                log.error("任务接收人信息不存在[{}]", taskReceiverId);
                return;
            }

            PushContext context = buildContext(taskReceiver);

            // 替换占位符
            replaceContent(context, true);

            // 检查消息有效性
            Pair<Long, CheckError> pair = checkContent(context);

            Long checkElapsedTime = pair.getLeft();
            context.setCheckElapsedTime(checkElapsedTime);
            CheckError checkError = pair.getRight();
            if (checkError != null) {
                savePushRecord(taskReceiver, context, checkError);  // 检查未通过，全部放弃推送
                for (Alarm alarm : alarms) {
                    if (alarm.supports(context, checkError)) {
                        alarm.alert(context, checkError);
                    }
                }
            } else {
                // 替换占位符
                replaceContent(context, false);

                checkError = send(context);
                savePushRecord(taskReceiver, context, checkError);
            }

            watch.stop();
            log.info("推送完成耗时{}秒[taskReceiverId:{}]", Formats.NUMBER_FORMAT.format(watch.getTotalTimeSeconds()), taskReceiver.getId());
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            throw new RuntimeException(e);
        }
    }

    /**
     * @return true - 全局检查不通过，不需要发送消息；false - 全局检查通过，需要在推送消息时进行消息过滤
     */
    private Pair<Long, CheckError> checkContent(PushContext context) {
        StopWatch watch = new StopWatch();
        watch.start();

        CheckError checkError = null;
        Set<String> checkerNames = Collections.emptySet();
        if (context.getTask().getReceiverType() == ReceiverType.ROOM) {
            RoomCategory roomCategory = context.getTask().getRoomCategory();
            if (StringUtils.hasText(roomCheckers)) {
                String[] maps = roomCheckers.split(";");
                checkerNames = Arrays.stream(maps)  // CustomerRoom:taskChecker,storeChecker
                        .map(it -> it.split(":"))   // CustomerRoom taskChecker,storeChecker
                        .filter(it -> RoomCategory.valueOf(it[0]) == roomCategory)
                        .flatMap(it -> Stream.of(it[1].split(",")))
                        .collect(Collectors.toSet());
            }
        } else {
            if (StringUtils.hasText(contactCheckers)) {
                String[] maps = contactCheckers.split(";");
                checkerNames = Arrays.stream(maps)
                        .flatMap(it -> Stream.of(it.split(",")))
                        .collect(Collectors.toSet());
            }
        }

        for (String checkerName : checkerNames) {
            Checker checker = checkers.get(checkerName);
            if (checker.supports(context)) {
                checkError = checker.check(context);
                if (checkError != null) {
                    log.info("任务明细检查不通过[taskId:{}, checkError:{}]", context.getTask().getId(), JSON.toJSONString(checkError));
                    break;
                }
            }
        }

        watch.stop();
        return Pair.of(watch.getTotalTimeMillis(), checkError);
    }

    private void replaceContent(PushContext context, boolean beforeCheck) {
        Task task = context.getTask();
        for (ContentGetter contentGetter : contentGetters) {
            if (contentGetter.beforeCheck() == beforeCheck) {
                contentGetter.replaceContent(context);
                log.info("任务明细内容替换已完成[taskId:{}, contentGetter:{}]", task.getId(), contentGetter.getClass().getSimpleName());
            }
        }
    }

    private PushContext buildContext(TaskReceiver taskReceiver) {
        Triple<Task, List<TaskItem>, Receiver> triple = getTaskAndReceiver(taskReceiver);
        Task task = triple.getLeft();
        List<TaskItem> taskItems = triple.getMiddle();  // 原始的item，推送时需要替换占位符
        Receiver receiver = triple.getRight();

        PushContext context = new PushContext();
        context.setTask(task);
        context.setTaskItems(taskItems);
        context.setReceiver(receiver);

        Robot robot = robotSelector.select(task, receiver);
        context.setRobot(robot);

        Long storeId = receiver.getStoreId();
        if (storeId != null) {
            Store store = storeClient.getStore(storeId);
            context.setStore(store);
        }
        storeId = task.getStoreId();
        if (storeId != null) {
            Store store = storeClient.getStore(storeId);
            context.setCutomStore(store);
        }
        return context;
    }

    private Triple<Task, List<TaskItem>, Receiver> getTaskAndReceiver(TaskReceiver taskReceiver) {
        Task task = taskMapper.selectById(taskReceiver.getTaskId());
        List<TaskItem> taskItems = getRandomTaskItems(task);
        Long storeId = taskReceiver.getStoreId();
        Store store = null;
        if (storeId != null) {
            store = storeClient.getStore(storeId);
        }
        Receiver receiver = taskReceiver.toReceiver(store);
        return Triple.of(task, taskItems, receiver);
    }

    private List<TaskItem> getRandomTaskItems(Task task) {
        Long taskId = task.getId();
        RAtomicLong pushedCount = redissonClient.getAtomicLong(RedisKeys.taskReceiverPushedCount(taskId, LocalDate.now()));
        long group = pushedCount.getAndIncrement() % task.getGroupCount();
        LambdaQueryWrapper<TaskItem> wrapper = new LambdaQueryWrapper<TaskItem>()
                .eq(TaskItem::getTaskId, taskId)
                .eq(TaskItem::getPsn, group);
        return taskItemMapper.selectList(wrapper);
    }

    private CheckError send(PushContext context) {
        addTrackData(context);
        SendChatMessagesRequest request = buildRequest(context);
        if (request == null) {
            return CheckError.ItemsNotValid;
        }
        SendChatMessagesResponse response = tanJingClient.execute(request);
        CheckError checkError = null;
        if (response.isSuccess()) {
            log.info("消息已发送[taskId:{}, request:{}, response:{}]",
                    context.getTask().getId(), JSON.toJSONString(request), JSON.toJSONString(response));
        } else {
            log.error("发送消息失败[taskId:{}, request:{}, response:{}]",
                    context.getTask().getId(), JSON.toJSONString(request), JSON.toJSONString(response));
            checkError = CheckError.UnknownError;
        }
        return checkError;
    }

    private void addTrackData(PushContext context) {
        Task task = context.getTask();
        Receiver receiver = context.getReceiver();
        List<TaskItem> taskItems = context.getTaskItems();
        for (TaskItem taskItem : taskItems) {
            if (taskItem.isMiniProgram()) {
                JSONObject content = taskItem.getContent();
                JSONObject trackData = new JSONObject();
                trackData.put("c", 10005);
                trackData.put("taskId", task.getId());
                trackData.put("receiverType", receiver.getType());
                trackData.put("receiverId", receiver.getId());
                if (context.getStore() != null) {
                    trackData.put("storeId", receiver.getStoreId());
                    trackData.put("storeCode", receiver.getStoreCode());
                } else if (context.getCutomStore() != null) {
                    Store defaultStore = context.getCutomStore();
                    trackData.put("storeId", defaultStore.getId());
                    trackData.put("storeCode", defaultStore.getCode());
                }
                trackData.put("skuSn", content.getString("skuSn"));
                if (!ObjectUtils.isEmpty(context.getTrackData())) {
                    trackData.putAll(context.getTrackData());
                }

                String pagePath = content.getString("pagePath");
                String newPagePath = UriComponentsBuilder.fromPath(pagePath)
                        .queryParam("p", trackData.toJSONString())
                        .replaceQueryParam("storeId", receiver.getStoreId())
                        .toUriString();
                content.put("pagePath", newPagePath);
            }
        }
    }

    @CheckReturnValue
    private SendChatMessagesRequest buildRequest(PushContext context) {
        Task task = context.getTask();
        List<TaskItem> taskItems = context.getTaskItems();
        Robot robot = context.getRobot();
        Receiver receiver = context.getReceiver();
        ReceiverType receiverType = context.getReceiver().getType();

        SendChatMessagesRequest request = switch (receiverType) {
            case CONTACT -> new SendPrivateChatMessagesRequest()
                    .setToWxSerialNo(receiver.getSerialNo());
            case ROOM -> new SendGroupChatMessagesRequest()
                    .setChatRoomSerialNo(receiver.getSerialNo());
        };

        List<PrivateChatMessage> data = taskItems.stream()
                .filter(it -> {
                    Integer errorCode = it.getErrorCode();
                    if (errorCode != null) {
                        log.info("任务明细无效[taskId:{}, taskItemId:{}, errorCode:{}, errorMessage:{}]",
                                task.getId(), it.getId(), errorCode, it.getErrorMessage());
                        return false;
                    }
                    return true;
                })
                .map(it -> {
                    PrivateChatMessage message = switch (receiverType) {
                        case CONTACT -> new PrivateChatMessage();
                        case ROOM -> new GroupChatMessage()
                                .setIsHit(0);
                    };
                    message.setMsgNum(task.getId())
                            .setMsgType(it.getMessageType().getCode());
                    return message;
                })
                .toList();
        if (ObjectUtils.isEmpty(data)) {
            return null;
        }

        request.setRobotSerialNo(robot.getSerialNo())
                .setRelaSerialNo(task.getId().toString())
                .setData(data);
        return request;
    }

    /**
     * 如果checkError不为null，表示发生了全局错误，将推送状态直接置为失败。
     */
    private void savePushRecord(TaskReceiver taskReceiver, PushContext context, CheckError checkError) {
        LocalDateTime now = LocalDateTime.now();
        PushRecord pushRecord = context.toPushRecord(idGenerator::next, now);
        pushRecord.setTaskReceiverId(taskReceiver.getId());
        List<PushRecordItem> pushRecordItems = context.toPushRecordItems(idGenerator::next, pushRecord.getId(), now);

        if (checkError != null) {
            ErrorSource errorSource = ErrorSource.LEO;
            for (PushRecordItem recordItem : pushRecordItems) {
                Integer errorCode = checkError.getErrorCode();
                recordItem.setErrorSource(errorSource);
                recordItem.setErrorCode(errorCode);
                recordItem.setErrorMessage("【" + errorSource.getName() + "】" + errorCode + ": " + checkError.getErrorMessage());
                recordItem.setStatus(SendStatus.FAIL);
            }
        }

        transactionTemplate.executeWithoutResult(_ -> {
            pushRecordMapper.insert(pushRecord);
            pushRecordItemMapper.insert(pushRecordItems);
        });
    }
}
